package com.zhisheng.sql.blink.stream.example;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.types.Row;

import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

public class RealtimeRecommendationJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        TableConfig config = tableEnv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1s");
        config.getConfiguration().set(TaskManagerOptions.NETWORK_MEMORY_FRACTION,0.4F);
        env.setParallelism(2);
        config.getConfiguration().set(TaskManagerOptions.MEMORY_SEGMENT_SIZE, new MemorySize(32768));
        config.getConfiguration().set(TaskManagerOptions.NETWORK_MEMORY_MIN,new MemorySize(1024 * 1024 * 1024L));
        config.getConfiguration().set(TaskManagerOptions.NETWORK_MEMORY_MAX,new MemorySize(1024 * 1024 * 1024L));

        // 1. 创建源表
        createSourceTables(tableEnv);

        // 2. 计算用户特征
        Table userFeatures = calculateUserFeatures(tableEnv);
        // 2.1 处理用户购买记录，使用Set存储
        DataStream<Row> ordersStream = tableEnv.toChangelogStream(
            tableEnv.sqlQuery("SELECT user_id, product_id, total_amount FROM orders")

        );
        DataStream<String> purchaseStream = ordersStream
            .map(row -> new UserPurchaseSetProcessor.OrderEvent(
                row.getField(0) != null ? (Integer) row.getField(0) : null,
                row.getField(1) != null ? (Integer) row.getField(1) : null,
                row.getField(2) != null ? (Double) row.getField(2) : null
            ))
            .keyBy(UserPurchaseSetProcessor.OrderEvent::getUserId)
            .process(new UserPurchaseSetProcessor());

        // 3. 计算用户相似度并生成推荐
        DataStream<Row> userFeaturesStream = tableEnv.toChangelogStream(userFeatures);
        //userFeaturesStream.print();
        DataStream<Tuple4<Integer, Integer, Double, List<String>>> recommendations =
                calculateSimilarityAndRecommend(userFeaturesStream);

        // 4. 输出推荐结果到Kafka
        createAndInsertRecommendations(tableEnv, recommendations);

        env.execute("Realtime Product Recommendation Job");
    }

    private static void createSourceTables(StreamTableEnvironment tableEnv) {
        // 用户表
        tableEnv.executeSql(
                "CREATE TABLE users (" +
                        "  user_id INT," +
                        "  username STRING," +
                        "  mobile STRING," +
                        "  PRIMARY KEY (user_id) NOT ENFORCED" +
                        ") WITH " +
                        "  ('connector' = 'jdbc',\n" +
                        "    'url' = 'jdbc:mysql://192.168.50.235:3306/test_sm?serverTimezone=Asia/Shanghai',\n" +
                        "    'table-name' = 'users',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = 'Cyx1018!',\n" +
                        "    'sink.buffer-flush.max-rows' = '1', -- 实时写入（每条立即插入）\n" +
                        "    'sink.max-retries' = '3'\n" +
                        "  )");


        // 产品表
        tableEnv.executeSql(
                "CREATE TABLE products (" +
                        "  product_id INT," +
                        "  product_name STRING," +
                        "  price DECIMAL(10, 2)," +
                        "  category STRING," +
                        "  created_at TIMESTAMP(3)," +
                        "  PRIMARY KEY (product_id) NOT ENFORCED" +
                        ") WITH " +
                        "  ('connector' = 'jdbc',\n" +
                        "    'url' = 'jdbc:mysql://192.168.50.235:3306/test_sm?serverTimezone=Asia/Shanghai',\n" +
                        "    'table-name' = 'products',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = 'Cyx1018!',\n" +
                        "    'sink.buffer-flush.max-rows' = '1', -- 实时写入（每条立即插入）\n" +
                        "    'sink.max-retries' = '3'\n" +
                        "  )");


        // 订单表
        tableEnv.executeSql(
                "CREATE TABLE orders (" +
                        "  order_id INT," +
                        "  user_id INT," +
                        "  product_id INT," +
                        "  quantity INT," +
                        "  total_amount DOUBLE" +
                         ") WITH ("
                        + "  'connector' = 'kafka',"
                        + "  'topic' = 'mysql-server.test_sm.orders ',"
                        + "  'properties.bootstrap.servers' = '192.168.50.235:9092,192.168.50.235:9093,192.168.50.235:9094',"
                        + "  'properties.group.id' = 'test_consumer',"
                        + "  'scan.startup.mode' = 'latest-offset',"
                        + "  'format' = 'debezium-json',"
                        + "  'debezium-json.schema-include' = 'true'"
                        + ")");
    }

    private static Table calculateUserFeatures(StreamTableEnvironment tableEnv) {
        // 1. 计算用户总消费
        Table userTotalSpending = tableEnv.sqlQuery(
                "SELECT " +
                        "    o.user_id, " +
                        "    SUM(o.total_amount) as total_spent, " +
                        "    COUNT(DISTINCT p.category) as category_count " +
                        "FROM orders o " +
                        "JOIN products p ON o.product_id = p.product_id " +
                        "GROUP BY o.user_id"
        );
        tableEnv.createTemporaryView("UserTotalSpending", userTotalSpending);
        // 2. 计算用户类别偏好
        Table userCategoryPreference = tableEnv.sqlQuery(
                "SELECT " +
                        "    o.user_id, " +
                        "    p.category, " +
                        "    COUNT(*) as category_count, " +
                        "    SUM(o.total_amount) as category_spent " +
                        "FROM orders o " +
                        "JOIN products p ON o.product_id = p.product_id " +
                        "GROUP BY o.user_id, p.category"
        );
        tableEnv.createTemporaryView("UserCategoryPreference", userCategoryPreference);

        // 3. 组合用户特征
        return tableEnv.sqlQuery(
                "SELECT " +
                        "    u.user_id, " +
                        "    u.username, " +
                        "    uts.total_spent as spending_level, " +
                        "    LISTAGG(DISTINCT ucp.category, ',') as preferred_categories " +
                        "FROM users u " +
                        "JOIN UserTotalSpending uts ON u.user_id = uts.user_id " +
                        "LEFT JOIN UserCategoryPreference ucp ON u.user_id = ucp.user_id " +
                        "GROUP BY u.user_id, u.username, uts.total_spent"
        );
    }

    private static DataStream<Tuple4<Integer, Integer, Double, List<String>>> calculateSimilarityAndRecommend(
            DataStream<Row> userFeatures) {

        return userFeatures
                .keyBy(row -> 1) // 全局keyBy以便后续计算相似度
                .window(TumblingProcessingTimeWindows.of(Time.seconds(30)))
                .apply(new UserSimilarityFunction());
    }

    private static void createAndInsertRecommendations(
            StreamTableEnvironment tableEnv,
            DataStream<Tuple4<Integer, Integer, Double, List<String>>> recommendations) {

        // 配置 JDBC Sink 将数据实时写入 MySQL
        String mysqlUrl = "jdbc:mysql://192.168.50.235:3306/test_sm";
        String mysqlUser = "root";
        String mysqlPassword = "Cyx1018!";

        // 将Tuple4转换为Row
        DataStream<Row> rowDataStream = recommendations.map(new MapFunction<Tuple4<Integer, Integer, Double, List<String>>, Row>() {
            @Override
            public Row map(Tuple4<Integer, Integer, Double, List<String>> value) throws Exception {
                return Row.of(
                    value.f0,  // user_id
                    value.f1,  // recommended_user_id
                    value.f2,  // similarity_score
                    String.join(",", value.f3),  // preferred_categories
                    new Timestamp(System.currentTimeMillis())  // update_time
                );
            }
        });

        // 设置 JDBC Sink
        SinkFunction<Row> jdbcSink = JdbcSink.sink(
                "INSERT INTO recommendation_results (user_id, recommended_user_id, similarity_score, preferred_categories, update_time) " +
                "VALUES (?, ?, ?, ?, ?) " +
                "ON DUPLICATE KEY UPDATE " +
                "recommended_user_id = VALUES(recommended_user_id), " +
                "similarity_score = VALUES(similarity_score), " +
                "preferred_categories = VALUES(preferred_categories), " +
                "update_time = VALUES(update_time)",
                (ps, row) -> {
                    ps.setInt(1, (Integer) row.getField(0));  // user_id
                    ps.setInt(2, (Integer) row.getField(1));  // recommended_user_id
                    ps.setDouble(3, (Double) row.getField(2));  // similarity_score
                    ps.setString(4, (String) row.getField(3));  // preferred_categories
                    ps.setTimestamp(5, (Timestamp) row.getField(4));  // update_time
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(1)  // 每条数据立即写入
                        .withMaxRetries(3)  // 失败重试3次
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(mysqlUrl)
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername(mysqlUser)
                        .withPassword(mysqlPassword)
                        .build()
        );

        // 添加sink
        rowDataStream.addSink(jdbcSink);
    }
}
